package main

import (
	"context"
	"fmt"
	"github.com/apache/rocketmq-client-go/v2"
	"github.com/apache/rocketmq-client-go/v2/consumer"
	"github.com/apache/rocketmq-client-go/v2/primitive"
	"github.com/apache/rocketmq-client-go/v2/rlog"
	"go_test/src/code.song.study/message/util"
	"time"
)

func init() {
	// RocketMQ 日志
	rlog.SetLogLevel("warn")
}

func main() {

	client, err := rocketmq.NewPushConsumer(
		consumer.WithGroupName(util.GetConsumerGroup()),
		consumer.WithNameServer(util.GetRocketNameSrv()),
		consumer.WithConsumeFromWhere(consumer.ConsumeFromLastOffset),
		consumer.WithConsumerModel(consumer.Clustering),
		consumer.WithConsumeMessageBatchMaxSize(1),
	)
	if err != nil {
		fmt.Println("consumer init error >>> ", err)
		return
	}

	initErr := client.Subscribe(util.TradeData, consumer.MessageSelector{}, func(ctx context.Context,
		msgs ...*primitive.MessageExt) (consumer.ConsumeResult, error) {
		for i := range msgs {
			msg := msgs[i]
			fmt.Printf("received trade data: %s \n", string(msg.Body))
		}
		return consumer.ConsumeSuccess, nil
	})

	client.Subscribe("trade_contract_order", consumer.MessageSelector{}, handContractOrder)
	if initErr != nil {
		fmt.Println("consumer sub error >>> ", initErr)
		return
	}
	err = client.Start()
	if err != nil {
		fmt.Println("consumer start error >>> ", err)
	}
	go shutdown()

	<-util.GetChan()
	//客户端自带的优雅退出
	client.Shutdown()
}

func shutdown() {
	fmt.Println("监听结果 >>>> ", time.Now().Format("2006-01-02 15:04:05"))
	time.Sleep(2 * time.Minute)
	fmt.Println("往通道写关闭信号 >>>>> ", time.Now().Format("2006-01-02 15:04:05"))
	util.GetChan() <- "quit"
}

// 处理合约订单
func handContractOrder(ctx context.Context, msgList ...*primitive.MessageExt) (consumer.ConsumeResult, error) {
	for _, msg := range msgList {
		fmt.Println("[trade_contract_order] --- order >>>>> ", string(msg.Body))
	}
	return consumer.ConsumeSuccess, nil
}
